package report

import java.util.Properties

import Configer.Config
import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import util.KpiPublic

//用广播变量的方式分析app分布
object AppKpiBro {
  def main(args: Array[String]): Unit = {
    //sparkContext
    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName(s"${this.getClass.getName}")
    conf.set("spark.serializer",Config.serializer)

    val sc = new SparkContext(conf)

    val dictLines = sc.textFile("C:\\Users\\44323\\Desktop\\资料PDF\\app_dict.txt")
    val values = dictLines.map(_.split("\t",-1)).filter(_.length>=5).map(arr=>(arr(4),arr(1))).collectAsMap()

    val value = sc.broadcast(values)

    val sQLContext = new SQLContext(sc)

    val dataFrame = sQLContext.read.parquet(Config.parquetPath)

    val resultRDD: RDD[(String, List[Double])] = dataFrame.map(row => {
      var appName = row.getAs[String]("appname")
      val appId = row.getAs[String]("appid")

      if (StringUtils.isEmpty(appName)) {
        if (StringUtils.isNotEmpty(appId)) {
          appName = value.value.getOrElse(appId, appId)
        } else {
          appName = "某app"
        }
      }
      (appName, KpiPublic.KpiPublic(row))
    }).reduceByKey((l1, l2) => l1 zip l2 map (t => t._1 + t._2))
    resultRDD
    val props = new Properties()
    props.setProperty("driver",Config.driver)
    props.setProperty("user",Config.user)
    props.setProperty("password",Config.password)

    import sQLContext.implicits._
    resultRDD.map(row=>(row._1,row._2(0),row._2(1),row._2(2),row._2(3),row._2(4),row._2(7),row._2(8),row._2(5),row._2(6)))
      .toDF("appname","adOldReq","adEffReq","adReq","adRtbReq","adSucReq","adshow","adclick","adadpReq","adCusReq")
      .write.jdbc(Config.url,Config.table,props)

    sc.stop()

  }
}
